﻿//========================================================= 
/**@file lmdb_event_svr.h
 * @brief 以lmdb事件方式进行通信的监控项
 * 
 * @date 2018-10-12   16:26:39
 * @author Zhyioo 
 * @version 1.0
 */ 
//--------------------------------------------------------- 
#ifndef _LIBZHOUYB_LMDB_EVENT_SVR_H_
#define _LIBZHOUYB_LMDB_EVENT_SVR_H_
//--------------------------------------------------------- 
#include "zabbix_converter.h"

#include "../lmdb/lmdb_event.h"
using zhou_yb::wrapper::lmdb_extractor::mutex_signal;
using zhou_yb::wrapper::lmdb_extractor::lmdb_string_converter;
using zhou_yb::wrapper::lmdb_extractor::lmdb_event_agent;
using zhou_yb::wrapper::lmdb_extractor::lmdb_event_server;

#include "../rapidjson/rapidjson_extrator.h"
using zhou_yb::wrapper::rapidjson_extractor::RapidJsonArgParser;

#include <boost/thread/thread_pool.hpp>
//--------------------------------------------------------- 
/// lmdb服务名称
#ifndef ZABBIX_LMDB_NAME
#   define ZABBIX_LMDB_NAME "LmdbZabbix"
#endif
//--------------------------------------------------------- 
namespace zhou_yb {
namespace wrapper {
namespace zabbix_extractor {
//--------------------------------------------------------- 
/**
 * @brief lmdb事件服务端
 *
 * @description 监听指定的事件,客户端将自己唯一的ID写到lmdb中,服务端根据ID创建与之对应的
 *
 * @date 2018-10-14 09:32:53
 */
class lmdb_event_svr
{
public:
    typedef mutex_signal signal_type;
    typedef converter_dispatcher::callback_type callback_type;
    typedef lmdb_event_agent<signal_type, lmdb_string_converter> agent_type;
    typedef lmdb_event_server<signal_type> server_type;

    static bool write_json(agent_type& agent, const IStringArguments& arg)
    {
        RapidJsonArgParser parser;
        ByteBuilder buff(64);
        parser.ToString(arg, buff);
        return agent.write(string(buff.GetString(), buff.GetLength()));
    }
    static bool read_json(agent_type& agent, IStringArguments& rlt, 
        uint timeoutMs = DEV_WAIT_TIMEOUT, 
        Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        string data;
        if(!agent.wait(data, timeoutMs, interrupter))
            return false;
        RapidJsonArgParser parser;
        parser.Parse(ByteArray(data.c_str(), data.length()), rlt);
        return true;
    }
protected:
    bool _isLastOK;
    BoolInterrupter _interrupter;
    list_t<agent_type> _clients;
    lmdb_event_server<signal_type> _server;
    converter_dispatcher _dispatcher;
    boost::basic_thread_pool _pool;

    void remove_agent(agent_type* agent)
    {
        list_t<agent_type>::iterator itr;
        for(itr = _clients.begin();itr != _clients.end(); ++itr)
        {
            if(agent == &(*itr))
            {
                _clients.erase(itr);
                break;
            }
        }
    }
    void async_callback(agent_type* agent, const string& cmd, bool isOK, const string& param, const string& rlt)
    {
        int err = isOK ? DeviceError::Success : DeviceError::OperatorErr;
        StringArguments r;
        r.Put("RLT", rlt);
        r.Put("CODE", ArgConvert::ToString<int>(err));

        write_json(*agent, r);
        agent->disconnect();
        remove_agent(agent);
    }
    void remote_wait_proc(agent_type* agent, uint timeoutMs, Ref<IInterrupter> interrupter)
    {
        StringArguments a;
        if(read_json(*agent, a, timeoutMs, interrupter))
        {
            string cmd = a["Command"].To<string>();
            string arg = a["Arg"].To<string>();
            callback_type callback = std::bind(&lmdb_event_svr::async_callback, this, agent,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4);
            int err = _dispatcher.async(cmd, arg, callback);
            if(err == DeviceError::Success) return;
        }
        remove_agent(agent);
    }
public:
    lmdb_event_svr() {}
    lmdb_event_svr(const char* dir, const char* svr, const char* lmdb_dir)
    {
        start(dir, svr, lmdb_dir);
    }
    virtual ~lmdb_event_svr() { stop(); }

    /**
     * @brief 开始服务
     * @date 2018-10-30 00:26
     * 
     * @param [in] dir driver存放目录
     * @param [in] svr lmdb服务名称
     * @param [in] lmdb_dir lmdb存放目录
     */
    int start(const char* dir, const char* svr, const char* lmdb_dir)
    {
        int err = _server.startup(svr, lmdb_dir);
        if(err == DeviceError::Success)
        {
            _isLastOK = true;
            _interrupter.Reset();
            _dispatcher.load(dir);
        }
        return err;
    }
    /// 返回加载的命令集
    inline size_t get(list_t<string>& cmds) const
    {
        return _dispatcher.get(cmds);
    }
    inline bool is_open() const
    {
        return _server.is_valid();
    }
    /**
     * @brief 等待单个客户端连接
     * @date 2018-10-30 00:27
     */
    int run(uint timeoutMs = DEV_WAIT_TIMEOUT, 
        Ref<IInterrupter> interrupter = Ref<IInterrupter>(), string* pName = NULL)
    {
        if(_isLastOK) _clients.push_back();
        int err = _server.accept(_clients.back(), timeoutMs, interrupter);
        _isLastOK = (err == DeviceError::Success);
        if(_isLastOK)
        {
            if(pName != NULL) (*pName) = _clients.back().getName();
            Ref<IInterrupter> subInterrupter = _interrupter;
            _pool.submit(std::bind(&lmdb_event_svr::remote_wait_proc, this, &_clients.back(), timeoutMs, subInterrupter));
        }
        return err;
    }
    void stop()
    {
        _dispatcher.stop();

        _interrupter.Interrupt();
        _pool.join();

        _isLastOK = true;
        _clients.clear();
        _server.teardown();
    }
};
//--------------------------------------------------------- 
/**
 * @brief lmdb监控项
 * @date 2018-10-29 23:26:00
 */
class lmdb_event_monitor
{
public:
    typedef string value_type;

    typedef mutex_signal signal_type;
    typedef lmdb_event_agent<signal_type, lmdb_string_converter> agent_type;

    static value_type put_err(int err)
    {
        return ArgConvert::ToString<int>(-err);
    }

    /**
     * @brief 执行一次监控任务
     * @date 2018-10-30 00:29
     * 
     * @param [in] svr lmdb服务名称
     * @param [in] lmdb_dir lmdb存放目录(数据交换)
     * @param [in] cmd 需要调用的监控项名称名称
     * @param [in] arg 调用监控项的参数
     * @param [out] rlt 调用监控项后获取到的数据
     * @param [in] timeoutMs [default:DEV_WAIT_TIMEOUT] 等待连接和数据的超时时间
     * @param [in] interrupter [default:null] 中断器
     */
    static int run(const char* svr, const char* lmdb_dir, 
        const char* cmd, const char* arg, string& rlt, 
        uint timeoutMs = DEV_WAIT_TIMEOUT, 
        Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        agent_type agent;
        int err = lmdb_event_server<signal_type>::connect(svr, lmdb_dir, agent, timeoutMs, interrupter);
        if(err != DeviceError::Success) return err;

        StringArguments a;
        StringArguments r;
        a.Put("Command", _strput(cmd));
        a.Put("Arg", _strput(arg));

        err = DeviceError::TransceiveErr;
        if(lmdb_event_svr::write_json(agent, a) && 
            lmdb_event_svr::read_json(agent, r, timeoutMs, interrupter))
        {
            err = r["CODE"].To<int>(DeviceError::Success);
            if(err == DeviceError::Success)
            {
                rlt = r["RLT"].To<string>();
            }
        }
        agent.disconnect();
        return err;
    }
    /**
     * @brief 命令行执行监控项
     * @date 2018-10-30 00:31:17
     *
     * argv[1] : lmdb_dir lmdb数据库目录
     * argv[2] : cmd 调用的监控命令
     * argv[*] : arg 监控命令的命令行参数
     */
    static value_type execute(int argc, const char* argv[])
    {
        if(argc < 1) return put_err(DeviceError::ArgLengthErr);
        int offset = 1;
        const char* lmdb_dir = argv[offset++];
        const char* cmd = argv[offset++];
        string arg = "";
        for(int i = offset; i < argc; ++i)
        {
            arg += argv[i];
            arg += ' ';
        }
        if(argc > offset) arg.pop_back();
        string rlt = "";
        int err = run(ZABBIX_LMDB_NAME, lmdb_dir, cmd, arg.c_str(), rlt);
        if(err == DeviceError::Success) return rlt;
        return put_err(err);
    }
};
//--------------------------------------------------------- 
} // namespace zabbix_extractor 
} // namespace wrapper
} // namespace zhou_yb 
//--------------------------------------------------------- 
#endif // _LIBZHOUYB_LMDB_EVENT_SVR_H_
//========================================================= 